package com.gis.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author LnnuUser
 * @create 2021-08-27-下午6:36
 */
object Spark24_RDD_Operator_Req {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc: SparkContext = new SparkContext(sparkConf)


    // 获得想要的数据，省份和广告
    val textRDD: RDD[String] = sc.textFile("datas/agent.log")
//    val lineRDD: RDD[Array[String]] = textRDD.map(_.split(" "))
//    val province = lineRDD.take(1)
//    val GG = lineRDD.take(4)


    // 将省份和广告作为key，value为1
    val mapRDD: RDD[((String, String), Int)] = textRDD.map(
      line => {
        val datas = line.split(" ")
        ((datas(1), datas(4)), 1)
      }
    )

    // 聚合为（（省份，广告），次数）
    val reduceRDD: RDD[((String, String), Int)] = mapRDD.reduceByKey(
      _+_
    )
//    reduceRDD.collect().foreach(println)
    // key打散为（省份，（广告，次数））
    val newMapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      case ((prv, ad), sum) => {
        (prv, (ad, sum))
      }
    }

    // 根据省份分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey()

    // 组内数据按照次数降序，取前三
    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        //倒序排序
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )

    resultRDD.collect().foreach(println)


    sc.stop()

  }

}
